/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.table.service.impl;

import static org.apache.flink.table.api.Expressions.$;

import java.util.concurrent.ExecutionException;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.lzfto.flink.demo.common.table.PhysicalBasicTable;
import org.lzfto.flink.demo.table.service.FlinkConnectorTestService;

/**
 * 从本地测试库的一个schema插入到另外一个
 * 192.168.1.246 test/1232018
 */
public class StandardJdbcConnectorTest  implements FlinkConnectorTestService {
    @Override
    public void run() throws Exception {
        //1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env); // 来自于bridge
        env.setParallelism(1);
      //2. 创建表源头表和目标表
        String sourceWith=PhysicalBasicTable.getSourceTable("student").withConnector("jdbc").build();
        String sourceTableDDL = """
                create table student(
                   student_id BIGINT,
                   name String,
                   gender String,
                   birth_date Date,
                   email String
                )
                """+sourceWith;
        tabEnv.executeSql(sourceTableDDL);

        //目标表
        //Flink 中 字段类型需要严格匹配，否则会出现许多奇怪的问题
        //很可惜内置的函数不能用于create table语法的字段定义部分
        String targetWith=PhysicalBasicTable.getTargetTable( "student").withConnector("jdbc").build();
        String targetTableDDL = """
                create table student_target(
                   student_id BIGINT,
                   name STRING,
                   gender STRING,
                   birth_date DATE,
                   email STRING
                )"""+targetWith;
        tabEnv.executeSql(targetTableDDL);

        //3.插入前，请清理目标表数据 
        //由于使用的是jdbc连接器，不支持删除操作，所以需要自行做一些其它处理
        //不过我们倒是建议那么做，因为更加高效

        //4.执行插入操作
        //主要2000年之后出生
        Table srcTable = tabEnv.sqlQuery("SELECT * FROM student WHERE birth_date >= DATE '2000-01-01'");
        //只要男生
        srcTable=srcTable.filter($("gender").isEqual("Male"));
        TableResult insResult=srcTable.executeInsert("student_target");
        try {
        	//插入前确保没有重复的数据，否则会报告失败
        	//如果想看成功的示例，参见  JdbcBatchTest
			insResult.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		} catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println("执行完成");
    }
}
